
package models.spider.actor;

import models.spider.ISpider;
import models.spider.config.Factory;
import play.Logger;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;
import akka.actor.Actor;
import akka.actor.ActorRef;
import akka.actor.EmptyLocalActorRef;
import akka.actor.Props;
import akka.actor.UntypedActorFactory;
import akka.pattern.Patterns;
import akka.routing.RouterRoutees;

/**
 * Interface with the actor system.
 * 
 * @author laurent@opprecht.info
 * 
 */
public class CrawlerService implements ISpider {

    public static CrawlerService get(final Factory config) {
        return new CrawlerService(config);
    }
    
    public static class CrawlerFactory implements UntypedActorFactory {

        private static final long              serialVersionUID = 1L;

        protected models.spider.config.Factory config           = null;

        CrawlerFactory(final models.spider.config.Factory config) {
            this.config = config;
        }

        @Override
        public Actor create() {
            return new Crawler(config);
        }

    }

    protected ActorRef crawler = null;

    protected Factory  config  = null;

    public CrawlerService(final Factory config, final String name) {
        this.config = config;
    }

    public CrawlerService(final Factory config) {
        this(config, "crawler");
    }

    public String getName() {
        return config.getName();
    }

    public Factory getConfig() {
        return config;
    }

    public ActorRef getCrawler() {
        if (crawler != null) {
            return crawler;
        }
        String name = getName();
        crawler = play.libs.Akka.system().actorFor("/user/" + name);
        if (crawler instanceof EmptyLocalActorRef) {
            Props crawlerProperties = new Props(new CrawlerFactory(config));
            // crawlerProperties.withCreator(new CrawlerFacory(config));
            crawler = play.libs.Akka.system().actorOf(crawlerProperties, name);
        }
        return crawler;
    }

    public void start(final int threadCount) {
        getCrawler().tell(new Crawler.Start(threadCount),null);
    }

    public void stop() {
        getCrawler().tell(new Crawler.Stop(),null);
    }

	@Override
    public void add(final String url) {
        getCrawler().tell(new Crawler.Add(url),null);
    }

    @Override
    public void refresh(final String url) {
        getCrawler().tell(new Crawler.Refresh(url),null);
    }

    public Status.Details status() {
    	
        Future<Object> future1 = Patterns.ask(getCrawler(), new Crawler.CountTh(), 100000);
        int i= 0;
        try {
        	Logger.info("Asking Routees");
            Duration wait_for = Duration.create(10, java.util.concurrent.TimeUnit.SECONDS);
            RouterRoutees routees = (RouterRoutees) Await.result(future1, wait_for);
            Logger.info("Counting");            
            i = routees.routees().size();
        } catch (Exception ex) {
        	if(!(ex instanceof NullPointerException)) Logger.error("Time out", ex);
        }
        Future<Object> future = Patterns.ask(getCrawler(), new Crawler.Info(), 100000);
        try {
            Duration wait_for = Duration.create(10, java.util.concurrent.TimeUnit.SECONDS);
            Status.Details result = (Status.Details) Await.result(future, wait_for);
            result.setThreads(i);
            return result;
        } catch (Exception ex) {
        	Status.Details result = new Status.Details();
        	result.setThreads(i);
            Logger.error("Time out", ex);
            return result;
        }

    }
}
